package com.fourth.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fourth.bean.TableProcess;
import com.fourth.common.GmallConfig;
import com.fourth.utils.DruidDSUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @Author CZQ
 * @Date 2022/8/18 19:38
 * @Version 1.0
 */
public class TableProcessFunction  extends BroadcastProcessFunction<JSONObject,String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> stateDescriptor ;

    private static DruidDataSource druidDataSource;

    // 空参构造器
    public TableProcessFunction() {
    }

    // 构造器 加入构造器的目的是为了把DimApp中的广播状态描述器通过参数的形式传进来给processBroadcastElement 中写数据到状态中使用
    //便不需要重新new了

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        //构建连接池
         druidDataSource = DruidDSUtil.createDataSource();
    }

    /**
     * 操作主流
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //1.获取广播状态数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);
        TableProcess tableProcess = broadcastState.get(value.getString("table"));
        String type = value.getString("type");

        //2.过滤数据 行
        //先过滤出不需要的事实表 得到主流和广播流中配置信息都共有的表  然后&& 后边的是在做整行过滤
        if (tableProcess !=null && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))){

            filterColumns(value.getJSONObject("data"),tableProcess.getSinkColumns());

            //3.补充sinkTable字段并输出  为什么要补充SinkTablle字段是因为未来数据是要写入到Phoenix中的，所以要有表名
            value.put("sinkTable",tableProcess.getSinkTable());
            out.collect(value);

        }
    }

    /**
     * 操作广播流
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {

        //1.获取并解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        TableProcess tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);

        //2.建表
        checkTable(tableProcess.getSinkTable(),
                tableProcess.getSinkColumns(),
                tableProcess.getSinkPk(),
                tableProcess.getSinkExtend());

        //3.将数据写入状态中
        String key = tableProcess.getSourceTable();
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);
        broadcastState.put(key,tableProcess);

    }

    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        if (sinkPk == null || sinkPk.equals("")){
            sinkPk = "id";
        }
        if (sinkExtend == null  ){
            // 前面的字符串拼接一个null会报错，但是拼接一个 “” 不会报错
                    sinkExtend = "";
        }

        //拼接SQL语句 create table if not exists db.tn(id varchar primary key,name,varchar,sex varchar) xxx
        StringBuilder createTableSQL = new StringBuilder("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)  //数据库.表名
                .append(".")
                .append(sinkTable)
                .append("(");
        String[] colums = sinkColumns.split(",");
        for (int i = 0; i < colums.length; i++) {
            String colum = colums[i];

            //判断当前字段是否为主键
            if (sinkPk.equals(colum)){
                createTableSQL.append(colum).append(" varchar primary key");
            }else {
                createTableSQL.append(colum).append(" varchar");
            }

            //判断当前是否为最后一个字段
            if (i < colums.length - 1){
                createTableSQL.append(",");

            }

        }
        //上述步骤已经拼接到了create table if not exists db.tn(id varchar primary key,name,varchar,sex varchar
        createTableSQL.append(")").append(sinkExtend);

        System.out.println("建表语句"+ createTableSQL);

        DruidPooledConnection connection = null;
        PreparedStatement preparedStatement = null;

        try {
            //获取连接
            connection  =  druidDataSource.getConnection();
            System.out.println(connection);

            //编译SQL
            preparedStatement =   connection.prepareStatement(createTableSQL.toString());
            //执行SQL
            preparedStatement.execute();

            //释放连接资源
            preparedStatement.close();
            connection.close();
        } catch (SQLException e) {
                 e.getErrorCode();
            //throw new RuntimeException("建表"+ sinkTable+"失败");
        } finally {
            if (preparedStatement !=null){
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null){
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }

    }

    //过滤数据 列  对于JSONObject 加工或者遍历的话，可以将其当做Map数据结构来处理
    private void filterColumns(JSONObject data, String sinkColumns) {
        //防止出现字符串中包含问题（tm_name 与 name）
        String[] columns = sinkColumns.split(","); //这里是数组 数组是没有contains的方法的
        //所以要转成集合
        List<String> columnList = Arrays.asList(columns);


        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));

    }
}
